package main

import (
	"crypto/sha256"
	"database/sql"
	"encoding/hex"
	"encoding/json"
	"fmt"
	"log"
	"strconv"
	"sync"
	"time"
)


type pbft struct {
	//节点信息
	node *node
	//每笔请求自增序号
	sequenceID int64
	//锁
	lock sync.Mutex
	//该笔消息是否已进行Commit广播
	isCommitBroadcast map[string]bool
	//该笔消息是否已对客户端进行Reply
	isReply map[string]bool
	//该检查点是否已达成共识
	isCheckPoint map[int64]bool
	//验证后的消息所在的区块链
	bTrfChain 	*TraBlockChain
	brTrfChain  *TraBlockChain
	bidChain   	*BidBlockChain
	//节点数据库地址
	db *sql.DB
	//当前视图
	view int
	//水线
	checkPoint int64
	//检查点更新间隔
	checkSize int64
	//更换视图时，用于阻拦其它消息的接收
	ViewChangingFlag bool
	//判断是否已经发送newView
	isNewView bool
	//用于存储不同节点发来的viewChange消息
	isViewChange map[int]string
	//存储viewChange中prepared的prePrepare消息
	//计算newView的prePrepare消息
	preparedMap map[int]map[int64]*prePrepare


}

func NewPBFT(nodeID int, addr string) *pbft {
	p := new(pbft)
	p.ViewChangingFlag = false
	p.isNewView = false
	p.node = NewNode(nodeID)
	p.node.addr = addr
	p.isCommitBroadcast = make(map[string]bool)
	p.isReply = make(map[string]bool)
	p.isCheckPoint = make(map[int64]bool)
	p.db = ConnectDB(nodeID)
	p.bTrfChain = NewTraBlockChain(p.db)
	p.brTrfChain = NewTraBlockChain(p.db)
	p.bidChain = NewBidBlockChain(p.db)
	p.sequenceID = QueryEventMaxSeq(p.db)
	p.checkSize = 100
	p.checkPoint = (p.sequenceID / p.checkSize) * p.checkSize
	p.view = 0
	p.isViewChange = make(map[int]string)
	p.preparedMap = make(map[int]map[int64]*prePrepare)

	//digest := sha256.Sum256([]byte("1"))
	//db := ConnectClientDB()
	//for j := 0; j < 100; j++ {
	//	for i := 0; i < 10; i++ {
	//		t := time.Now().UnixNano()
	//		p.node.RsaSignWithSha256(digest[:], p.node.rsaPrivKey)
	//		//log.Println(time.Now().Nanosecond() - t)
	//		InsertTime1(db, time.Now().UnixNano()-t, 0)
	//	}
	//	time.Sleep(1000 * time.Millisecond)
	//}
	return p
}

//消息类型分类
func (p *pbft) handleRequest(data []byte) {
	//切割消息，根据消息命令调用不同的功能
	cmd, content := splitMessage(data)
	if p.ViewChangingFlag == false {
		switch command(cmd) {
		case cRequest:
			p.handleClientRequest(content)
		case cPrePrepare:
			p.handlePrePrepare(content)
		case cPrepare:
			p.handlePrepare(content)
		case cCommit:
			p.handleCommit(content)
		case cQuery:
			p.handleClientQuery(content)
		}
	}
	switch command(cmd) {
	case cCheckPoint:
		p.handleCheckPoint(content)
	case cViewChange:
		p.handleViewChange(content)
	case cNewView:
		p.handleNewView(content)
	}
	//log.Println(cmd)
}

//处理客户端请求
func (p *pbft) handleClientRequest(content []byte) {
	//使用json解析出Request结构体
	r := new(Request)
	err := json.Unmarshal(content, r)
	if err != nil {
		log.Panic(err)
	}
	digest := getDigest(*r)
	digestByte,_ := hex.DecodeString(r.Digest)
	//todo
	//由于在一台主机上允许，所以MAC一致，使用Addr索引。
	//后期可以改成MAC索引
	clientPubKey := QueryBidPubKeyByAddr(p.db, r.ClientAddr)

	if r.Operation != oNewNode && !p.node.RsaVerySignWithSha256(digestByte, r.Sign, clientPubKey){
		fmt.Println("签名有误，不接收该Request")
	}else if p.node.nodeID == p.view % nodeCount{
		fmt.Println("主节点已接收到客户端发来的request ...")

		num := ExistEventDigestAndView(p.db, digest, strconv.Itoa(p.view))
		if num == 0 {//检查是否收到过相同请求，重复请求不理睬
			//验证签名
			//添加信息序号
			p.sequenceIDAdd()
			//获取消息摘要
			fmt.Println("已将request存入临时消息池")
			//存入临时消息池
			//TODO
			//当请求多的时候，将请求存入缓冲池client_buffer，再从中获取event信息
			InsertEvent(p.db, strconv.FormatInt(p.sequenceID, 10), strconv.Itoa(p.view),
				r.Content, r.Digest, digest, strconv.FormatInt(r.Timestamp, 10),
				string(r.Operation), r.ClientAddr)

			//主节点对消息摘要进行签名
			digestByte, _ := hex.DecodeString(digest)
			signInfo := p.node.RsaSignWithSha256(digestByte, p.node.rsaPrivKey)
			//拼接成PrePrepare，准备发往follower节点
			pp := PrePrepare{p.node.nodeID, p.sequenceID, digest}
			ppR := PrePrepareRequest{pp, *r, signInfo}
			b, err := json.Marshal(ppR)
			if err != nil {
				log.Panic(err)
			}
			fmt.Println("正在向其他节点进行进行PrePrepare广播 ...")
			//进行PrePrepare广播
			p.broadcast(cPrePrepare, b)
			fmt.Println("PrePrepare广播完成")
		}
	}else{
		fmt.Println("副节点已接收到客户端发来的request ...")
		//1、查询数据库是否有对应消息
		seq := QueryBTrfIDByHash(p.db,r.Digest,r.ClientAddr)
		//回复客户端
		if seq != 0 {
			fmt.Println("正在返回已完成请求信息...")
			cc, err := QueryOneEventByDigest(p.db, r.Digest)
			if err != nil {
				fmt.Println(err)
			}
			digestByte, _ := hex.DecodeString(r.Digest)
			//将消息信息，提交到本地消息池中！
			sign := p.node.RsaSignWithSha256(digestByte, p.node.rsaPrivKey)
			var res Response
			res.SequenceID = seq
			rep := Reply{p.view, cc.timeStamp, cc.clientAddress,
				p.node.nodeID, res, sign}

			fmt.Println("正在reply客户端 ...")
			//tcpDial(content, cc.clientAddress)
			err = p.handleReplyKind(&rep)
			if err != nil {
				log.Panic(err)
			}
			p.isReply[r.Digest] = true
			fmt.Println("reply完毕")
		}else{
			//2、向主节点发送请求
			mainNode := strconv.Itoa(p.view % nodeCount)
			log.Printf("正在向主节点%s发送request",mainNode)
			reReq := jointMessage(cRequest,content)
			go tcpDial(reReq, nodeTable[mainNode])
			//3、启动计时器
			go p.simpleTimer(digest)
			//4、计时器到期判断是否更改视图
		}


	}
}


//处理预准备消息
func (p *pbft) handlePrePrepare(content []byte) {
	fmt.Println("本节点已接收到主节点发来的PrePrepare ...")
	//使用json解析出PrePrepare结构体
	pp := new(PrePrepareRequest)
	err := json.Unmarshal(content, pp)
	if err != nil {
		log.Panic(err)
	}
	//获取主节点的公钥，用于数字签名验证
	primaryNodePubKey := p.node.getPubKey(p.view % nodeCount)
	digestByte, _ := hex.DecodeString(pp.PrePrepareMessage.Digest)
	if digest := getDigest(pp.RequestMessage); digest != pp.PrePrepareMessage.Digest {
		fmt.Println("信息摘要对不上，拒绝进行prepare广播")
	} else if !p.node.RsaVerySignWithSha256(digestByte, pp.Sign, primaryNodePubKey) {
		fmt.Println("主节点签名验证失败！,拒绝进行prepare广播")
	} else if pp.PrePrepareMessage.View != p.view { //视图编号一致
		fmt.Println("视图编号对不上，拒绝进行prepare广播")
	} else if digest := QueryEventDigestByID(p.db, pp.PrePrepareMessage.SequenceID, pp.PrePrepareMessage.View);
	digest != "" && digest != pp.PrePrepareMessage.Digest {//从未在视图v中接受过序号为n但是摘要d不同的消息m
		fmt.Println("消息序号已存在，拒绝进行prepare广播")
	} else if (pp.PrePrepareMessage.SequenceID > p.checkPoint + 2 * p.checkSize) ||
		(pp.PrePrepareMessage.SequenceID < p.checkPoint) {//水线
		fmt.Println(p.checkPoint)
		fmt.Println(pp.PrePrepareMessage.SequenceID)
		fmt.Println("消息序号超出水线！,拒绝进行prepare广播")
	} else {
		//序号赋值
		p.sequenceID = pp.PrePrepareMessage.SequenceID
		//信息存入临时消息池
		fmt.Println("已将消息存入临时节点池")
		InsertEvent(p.db, strconv.FormatInt(pp.PrePrepareMessage.SequenceID,10),
			strconv.Itoa(pp.PrePrepareMessage.View), pp.RequestMessage.Content,
			pp.RequestMessage.Digest, pp.PrePrepareMessage.Digest,
			strconv.FormatInt(pp.RequestMessage.Timestamp, 10),
			string(pp.RequestMessage.Operation), pp.RequestMessage.ClientAddr)
		reqMsg, _ := json.Marshal(pp.RequestMessage)
		InsertPrePrepare(p.db, strconv.FormatInt(pp.PrePrepareMessage.SequenceID,10),
			strconv.Itoa(pp.PrePrepareMessage.View), pp.PrePrepareMessage.Digest, string(reqMsg))
		sign := p.node.RsaSignWithSha256(digestByte, p.node.rsaPrivKey)
		//拼接成Prepare
		pre := Prepare{pp.PrePrepareMessage.View, pp.PrePrepareMessage.SequenceID,
			pp.PrePrepareMessage.Digest, p.node.nodeID, sign}
		bPre, err := json.Marshal(pre)
		if err != nil {
			log.Panic(err)
		}
		InsertPrepare(p.db, strconv.Itoa(p.node.nodeID),
			strconv.FormatInt(pp.PrePrepareMessage.SequenceID,10),
			strconv.Itoa(pp.PrePrepareMessage.View), pp.PrePrepareMessage.Digest)
		//进行准备阶段的广播
		fmt.Println("正在进行Prepare广播 ...")
		p.broadcast(cPrepare, bPre)
		fmt.Println("Prepare广播完成")
	}
}

//处理准备消息
func (p *pbft) handlePrepare(content []byte) {
	//使用json解析出Prepare结构体
	pre := new(Prepare)
	err := json.Unmarshal(content, pre)
	if err != nil {
		log.Panic(err)
	}
	fmt.Printf("本节点已接收到%d节点发来的Prepare ... \n", pre.NodeID)
	//获取消息源节点的公钥，用于数字签名验证
	MessageNodePubKey := p.node.getPubKey(pre.NodeID)
	digestByte, _ := hex.DecodeString(pre.Digest)
	//签名、水线、视图
	if a := ExistEventDigest(p.db, pre.Digest); a == 0{
		fmt.Println("当前临时消息池无此摘要，拒绝执行commit广播")
	} else if p.view != pre.View{
		fmt.Println("视图序号对不上，拒绝执行commit广播")
	} else if (pre.SequenceID > p.checkPoint + 2 * p.checkSize) || (pre.SequenceID < p.checkPoint) {
		fmt.Println("消息序号超出水线！,拒绝进行commit广播")
		fmt.Println(pre.SequenceID)
		fmt.Println(p.checkPoint)
	} else if !p.node.RsaVerySignWithSha256(digestByte, pre.Sign, MessageNodePubKey) {
		fmt.Println("节点签名验证失败！,拒绝执行commit广播")
	} else if !p.isCommitBroadcast[pre.Digest]{ //未曾进行过Commit广播
		InsertPrepare(p.db, strconv.Itoa(pre.NodeID),
			strconv.FormatInt(pre.SequenceID, 10),
			strconv.Itoa(pre.View), pre.Digest)
		count := QueryRowsPrepare(p.db, pre.SequenceID)
		specifiedCount := nodeCount / 3 * 2
		p.lock.Lock()
		//获取消息源节点的公钥，用于数字签名验证
		if count >= specifiedCount {
			fmt.Println("本节点已收到至少2f个节点(包括本地节点)发来的Prepare信息 ...")
			//存储prepared_table
			Prepared(p.db, pre.SequenceID)
			//节点使用私钥对其签名
			sign := p.node.RsaSignWithSha256(digestByte, p.node.rsaPrivKey)
			c := Commit{ pre.View, pre.SequenceID, pre.Digest,
				p.node.nodeID, sign}
			bc, err := json.Marshal(c)
			if err != nil {
				log.Panic(err)
			}
			//进行提交信息的广播
			fmt.Println("正在进行commit广播")
			p.broadcast(cCommit, bc)
			p.isCommitBroadcast[pre.Digest] = true
			//DeletePrepare(p.db, pre.SequenceID)
			fmt.Println("commit广播完成")
		}
		p.lock.Unlock()
	}
}

//处理提交确认消息
func (p *pbft) handleCommit(content []byte) {
	//使用json解析出Commit结构体
	c := new(Commit)
	err := json.Unmarshal(content, c)
	if err != nil {
		log.Panic(err)
	}
	fmt.Printf("本节点已接收到%d节点发来的Commit ... \n", c.NodeID)
	//获取消息源节点的公钥，用于数字签名验证
	MessageNodePubKey := p.node.getPubKey(c.NodeID)
	digestByte, _ := hex.DecodeString(c.Digest)
	//签名水线视图
	if a := QueryRowsPrepareByDigest(p.db, c.Digest); a == 0{
		fmt.Println("当前prepare池无此摘要，拒绝将信息持久化到本地消息池")
	} else if p.view != c.View {
		fmt.Println("视图序号对不上，拒绝将信息持久化到本地消息池")
	} else if p.sequenceID != c.SequenceID {
		fmt.Println("消息序号对不上，拒绝将信息持久化到本地消息池")
	} else if (c.SequenceID > p.checkPoint + 2 * p.checkSize) || (c.SequenceID < p.checkPoint) {
		fmt.Println(c.SequenceID)
		fmt.Println("消息序号超出水线！,拒绝将信息持久化到本地消息池")
	} else if !p.node.RsaVerySignWithSha256(digestByte, c.Sign, MessageNodePubKey) {
		fmt.Println("节点签名验证失败！,拒绝将信息持久化到本地消息池")
	} else if !p.isReply[c.Digest]{//没有发送过REPLY
		InsertCommit(p.db, strconv.Itoa(c.NodeID),
			strconv.FormatInt(c.SequenceID, 10),
			strconv.Itoa(p.view), c.Digest)
		count := QueryRowsCommit(p.db, c.Digest)
		//如果节点至少收到了2f+1个commit消息（包括自己）,并且节点没有回复过,并且已进行过commit广播，则提交信息至本地消息池，并reply成功标志至客户端！
		p.lock.Lock()
		if count >= nodeCount/3*2 && p.isCommitBroadcast[c.Digest] {
			fmt.Println("本节点已收到至少2f + 1 个节点(包括本地节点)发来的Commit信息 ...")
			cc, err := QueryOneEventByDigest(p.db, c.Digest)
			if err != nil{
				fmt.Println(err)
			}
			//将消息信息，提交到本地消息池中！
			sign := p.node.RsaSignWithSha256(digestByte, p.node.rsaPrivKey)
			var res Response
			res.SequenceID = c.SequenceID
			r := Reply{c.View, cc.timeStamp, cc.clientAddress,
				p.node.nodeID, res, sign}

			fmt.Println("正在reply客户端 ...")
			//tcpDial(content, cc.clientAddress)
			err = p.handleReplyKind(&r)
			if err != nil{
				log.Panic(err)
			}
			p.isReply[c.Digest] = true
			fmt.Println("reply完毕")
			if c.SequenceID >= p.checkPoint + p.checkSize{
				p.checkPointMessage(c.SequenceID)
			}
		}
		p.lock.Unlock()
	}
}

//处理客户端发来的请求
func (p *pbft) handleReplyKind(r *Reply)(err error){
	e, err := QueryOneEventByID(p.db, r.ResponseMessage.SequenceID)
	if err != nil{
		log.Panic(err)
	}
	switch e.operation {
	case oNewNode:
		err = p.handleNewNode(&e, r)
	case oNewTra:
		err = p.handleNewTra(&e, r)
	case oVerTra:
		p.handleVerTra(&e, r)
	case oViewChange:
		//p.handleViewChange(e)
	}
	return
}

//向除自己外的其他节点进行广播
func (p *pbft) broadcast(cmd command, content []byte) {
	for i := range nodeTable {
		if i == strconv.Itoa(p.node.nodeID) {
			continue
		}
		message := jointMessage(cmd, content)
		go tcpDial(message, nodeTable[i])
	}
}

//序号累加
func (p *pbft) sequenceIDAdd() {
	p.lock.Lock()
	p.sequenceID++
	p.lock.Unlock()
}

//处理客户端的查询请求
func (p *pbft) handleClientQuery(content []byte) {
	//log.Println()
	q := new(Query)
	err := json.Unmarshal(content, q)
	if err != nil{
		log.Println(err)
	}
	msgHash := QueryBTrfBySeq(p.db,q.SequenceID)
	q.QueryReply = msgHash
	digestByte, _ := hex.DecodeString(strconv.FormatInt(q.SequenceID,10))
	q.Sign = p.node.RsaSignWithSha256(digestByte, p.node.rsaPrivKey)
	rContent,err := json.Marshal(q)

	message := jointMessage(cQuery, rContent)
	tcpDial(message,q.ClientAddr)
}

//处理新加入节点的reply包
func (p *pbft) handleNewNode(e *event,r *Reply) (err error) {
	//生成区块，上链
	cert := new(Cert)
	err = json.Unmarshal([]byte(e.message), cert)
	if err != nil{
		log.Panic(err)
		return
	}
	b := NewBidBlock(e.eventId, cert.Mac, cert.Addr, string(cert.PublicKey))
	b.Seal()
	p.bidChain.AddBlock(p.db,b)
	r.ResponseMessage.Type = oNewNode
	r.ResponseMessage.NodeCount = nodeCount
	r.ResponseMessage.NodeAddr = p.node.addr
	br,err := json.Marshal(r)
	if err != nil{
		return
	}
	content := jointMessage(cReply, br)
	tcpDial(content, e.clientAddress)
	return

	//返回nodeCount
	//返回nodeID、 nodeAddr
}

func (p *pbft) handleNewTra(e *event,r *Reply) (err error){
	b := NewTraBlock(e.msgDigest, r.ClientAddress, "", e.eventId)
	b.Seal()
	p.bTrfChain.AddBlock(p.db,b)
	r.ResponseMessage.Type = oNewTra
	r.ResponseMessage.Response = QueryBTrfHashByEventId(p.db, e.eventId)
	br,err := json.Marshal(r)
	if err != nil{
		return
	}
	content := jointMessage(cReply, br)
	tcpDial(content, e.clientAddress)
	return
}

//处理新业务的reply包
func (p *pbft) handleVerTra(e *event,r *Reply) {
	fmt.Println("VerTra:")
	//a.检查checked
	log.Println(e.message)
	checked, message := VeriTail([]byte(e.message))
	if checked{
		//检查btrf是否存在
		eventId, err := strconv.ParseInt(string(message),10,64)
		if err != nil{
			log.Println(err)
		}
		brt, err := QueryOneBTrf(p.db,eventId)
		if err!= nil{
			log.Println(err)
			//log.Println(eventId)
		}
		//b.trf(无dstAddr)->rtrf(从trf获取srcAddr)
		b := NewTraBlock(e.msgDigest, brt.srcAddress, r.ClientAddress, e.eventId)
		b.Seal()
		p.brTrfChain.AddBrTrfBlock(p.db,b)
		r.ResponseMessage.Type = oVerTra
		r.ResponseMessage.Response = "true"
		br,err := json.Marshal(r)
		if err != nil{
			return
		}
		content := jointMessage(cReply, br)
		tcpDial(content, e.clientAddress)
	}
	return
}

func (p *pbft) checkPointMessage(id int64) {
	fmt.Println("到达检查点，准备进行checkPoint共识")
	var cp CheckPoint
	cp.SequenceID = id
	cp.NodeID = p.node.nodeID
	//当前服务状态摘要
	digest := ""
	if id > p.checkSize{
		digest,_ = QueryCheckDigestByID(p.db, id)
	}
	digest = digest + NowBidHash(p.db, id) + NowBTrfHash(p.db, id) + NowBrTrfHash(p.db, id)
	cp.Digest = p.node.msgDigest(digest)
	digestByte, _ := hex.DecodeString(cp.Digest)
	//签名
	cp.Sign = p.node.RsaSignWithSha256(digestByte, p.node.rsaPrivKey)
	br,err := json.Marshal(cp)
	if err != nil{
		return
	}
	InsertCheckPoint(p.db, cp.SequenceID, cp.NodeID, cp.Digest)
	fmt.Println("正在进行checkPoint广播......")
	p.broadcast(cCheckPoint, br)
	fmt.Println("checkPoint广播完成")
}

func (p *pbft) handleCheckPoint(content []byte) {
	fmt.Println("收到checkPoint检查点")
	cp := new(CheckPoint)
	err := json.Unmarshal(content, cp)
	if err != nil{
		log.Panic(err)
	}
	digestByte, _ := hex.DecodeString(cp.Digest)
	MessageNodePubKey := p.node.getPubKey(cp.NodeID)
	if p.node.RsaVerySignWithSha256(digestByte, cp.Sign, MessageNodePubKey){
		InsertCheckPoint(p.db,cp.SequenceID,cp.NodeID,cp.Digest)
		//所发来的checkpoint是否有2f+1个相同digest
		a := QueryRowsCheckPoint(p.db, cp.SequenceID, cp.Digest)
		//fmt.Println(p.isCheckPoint)
		if a > nodeCount / 3 * 2 && p.isCheckPoint[p.checkPoint] != true{
			DeleteCheckPoint(p.db,cp.SequenceID)
			DeleteCommit(p.db,cp.SequenceID)
			DeletePrepare(p.db, cp.SequenceID)
			DeletePrePrepare(p.db, cp.SequenceID)
			DeleteEvent(p.db, cp.SequenceID)
			p.isCheckPoint[p.checkPoint] = true
			p.checkPoint = cp.SequenceID
			//log.Println(p.checkPoint)
			fmt.Println("checkPoint已达成共识")
		}
	}
}

func (p *pbft) simpleTimer(digest string) {
	time.Sleep(time.Second * 2)
	if p.isReply[digest] == true{
		return
	}else{

		fmt.Println("未收到主节点消息，将发送view-change")
		//log.Println(digest)
		//log.Println(p.isReply)
		p.ViewChange()
	}
}

//<VIEW-CHANGE,v+1,n,C,P,i>
func (p *pbft) ViewChange() {
	//1、阻塞所有其他数据
	p.ViewChangingFlag = true
	p.view++
	//2、n:最近一个检查点序号，从checkpoint_table获取
	vc := new(ViewChange)
	vc.View = p.view
	vc.NodeID = p.node.nodeID
	vc.SequenceID = NowCheckPoint(p.db)
	//3、组装CP
	//3-1数据库取出至少2f+1个相同checkpoint
	vc.CheckPointSet = GetCheckPointString(p.db)
	//3-1取出pre-prepare
	vc.PrePrepareSet = GetPPreAndPreString(p.db)
	//4、广播
	bvc, err := json.Marshal(vc)
	if err != nil{
		log.Panic(err)
	}
	fmt.Println("正在广播view-change包......")
	p.broadcast(cViewChange, bvc)
	fmt.Println("view-change广播完成")
}

//处理view-change，发送new-view
func (p *pbft) handleViewChange(content []byte) {
	if p.node.nodeID != p.view % nodeCount{

	}else{
		fmt.Println("主节点收到ViewChange:")
		//1、解码结构体
		vc := new(ViewChange)
		err := json.Unmarshal(content, vc)
		if err!= nil{
			log.Panic(err)
		}

		//2、存入view_change,用于组装new_view
		p.isViewChange[vc.NodeID] = string(content)
		//分解存入prepre/pre
		//3、获取当前check_point
		if p.checkPoint < vc.SequenceID{
			p.checkViewChangePoint(vc)
		}

		//4、处理pre-prepare
		//分解pre-prepare结构
		p.checkViewChangePre(vc.PrePrepareSet)

		//5、检查收到数量足够与否->是
		count := 0
		for range p.isViewChange{
			count++
		}
		//收到2f个view_change并且没发送过该消息
		if count > nodeCount * 2 / 3 - 1 && p.isNewView == false{
			//一个小问题，changeView到的太快，所以isNewVIEW还没来得及改变,会重复发送
			//等逻辑加入以后再测试
			p.isNewView = true

			//清空并重新存入pre-prepare数据
			DeletePrePrepare(p.db, p.checkPoint + p.checkSize * 2)
			for _,onePrePrepare := range p.preparedMap[p.view]{
				InsertPrePrepare(p.db, strconv.FormatInt(onePrePrepare.EventId, 10),
					strconv.Itoa(p.view), onePrePrepare.Digest, onePrePrepare.Message)
			}

			nv := new(NewView)
			nv.View = p.view
			//change-view->V
			vcByte, err := json.Marshal(p.isViewChange)
			if err != nil{
				log.Println(err)
			}
			nv.ViewChangeSet = string(vcByte)
			//获取最大序列号(可省略)
			if max := MaxPrePrepare(p.db); p.sequenceID < max{
				p.sequenceID = max
			}
			//对于pre-prepare表生成pre-prepare请求集合
			if p.sequenceID > p.checkPoint{
				nv.PrePrepareSet = GetPrePrepareString(p.db)
			}else{//如果相等，需要返回空
				p.sequenceID = p.checkPoint
				preMsg := new(PrePrepare)
				preMsg.View = p.view
				preMsg.SequenceID = p.sequenceID
				prepareSet := make(map[int]string)
				pp,err := json.Marshal(preMsg)
				if err != nil{
					log.Println(err)
				}
				prepareSet[p.node.nodeID] = string(pp)
				pps,err := json.Marshal(prepareSet)
				if err != nil{
					log.Println(err)
				}
				nv.PrePrepareSet = string(pps)
			}
			//主节点对pps签名
			vcHash := sha256.Sum256([]byte(nv.ViewChangeSet))
			ppHash := sha256.Sum256([]byte(nv.PrePrepareSet))
			//进行十六进制字符串编码
			//digest := hex.EncodeToString(vchash[:])+hex.EncodeToString(pphash[:])
			digestByte := append(vcHash[:],ppHash[:]...)
			nv.Sign = p.node.RsaSignWithSha256(digestByte, p.node.rsaPrivKey)
			//发送new-view
			content,err := json.Marshal(nv)
			if err!= nil {
				log.Panic(err)
			}
			fmt.Println("正在广播newView......")
			p.broadcast(cNewView, content)
			p.isNewView = false
			p.ViewChangingFlag = false
			fmt.Println("newView广播完成")
			//垃圾清理
			DeleteCheckPoint(p.db,p.checkPoint)
			DeletePrePrepare(p.db, p.checkPoint)
			fmt.Println("新任主节点垃圾清理完成")
		}
	}
}

func (p *pbft) checkViewChangePoint(vc *ViewChange){
	//3-1、获取，更新check_point
	checkString := make(map[int]string)
	err := json.Unmarshal([]byte(vc.CheckPointSet), &checkString)
	if err != nil{
		log.Println(err)
	}

	//3-2、存入checkpoint
	oneCheckPoint := new(checkPoint)
	for _,i := range checkString{
		err := json.Unmarshal([]byte(i), oneCheckPoint)
		if err != nil{
			log.Println("err :",err)
		}
		InsertCheckPoint(p.db, oneCheckPoint.EventId,
			oneCheckPoint.NodeId, oneCheckPoint.Digest)
	}

	//3-3、验证checkPoint,并获取最小的消息序列号
	a := QueryRowsCheckPoint(p.db, vc.SequenceID, oneCheckPoint.Digest)
	if a > nodeCount / 3 * 2 && p.isCheckPoint[p.checkPoint] != true{
		p.checkPoint = vc.SequenceID
	}
}

func (p *pbft)checkViewChangePre(ppSet string){
	//解码prePrepareSet
	pPreString := make(map[string]string)
	err := json.Unmarshal([]byte(ppSet), &pPreString)
	if err != nil{
		log.Println(err)
	}
	//遍历每一个prepared
	for pPre, pre := range pPreString{
		//获取pre-prepare
		onePrePrepare := new(prePrepare)
		err := json.Unmarshal([]byte(pPre), onePrePrepare)
		if err != nil{
			log.Println("err :",err)
		}
		//获取prepare集合
		preSet := make(map[int]string)
		err = json.Unmarshal([]byte(pre), &preSet)
		if err != nil{
			log.Println("err :",err)
		}
		//记录当前digest存在的nodeId
		prepared := make(map[string]map[int]bool)
		//获取每一个prepare
		for nodeId, prePareMessage := range preSet{
			//解码prepare
			onePrepare := new(prepare)
			err := json.Unmarshal([]byte(prePareMessage), onePrepare)
			if err != nil{
				log.Println("err :",err)
			}
			//记录prepare消息
			if prepared[onePrePrepare.Digest] == nil{
				prepared[onePrePrepare.Digest] = make(map[int]bool)
			}
			prepared[onePrepare.Digest][nodeId] = true
		}
		//验证prepared
		preparedCount := 0
		for range prepared[onePrePrepare.Digest]{
			preparedCount++
		}
		//验证成功，存入pre-prepare表，等待发送
		if preparedCount > nodeCount * 2 / 3 - 1{
			//存在性检查可以加入，存储之前判断是否存在
			//存入pre-prepare表
			/*
				存入pre-prepareMap好处：
				1、为了避免反复存入数据库时间长？
				2、为了避免发送pre-parepared 到时候发送数据库中未验证的pre-prepare
					2-1、可以存入之前清空数据库
				3、可以避免数据已经发送，但后续来的数据中有未发送的数据入库，造成数据重复冲突
					3-1、客户端没有收到消息会重复发送，时间戳不同不会计入
			*/
			if p.preparedMap[p.view] == nil{
				p.preparedMap[p.view] = make(map[int64]*prePrepare)
			}
			p.preparedMap[p.view][onePrePrepare.EventId] = onePrePrepare
			//InsertPrePrepare(p.db, strconv.FormatInt(onePrePrepare.EventId, 10),
			//	strconv.Itoa(p.view), onePrePrepare.Digest, onePrePrepare.Message)
		}
	}

}

func (p *pbft) handleNewView(content []byte) {
	nv := new(NewView)
	err := json.Unmarshal(content, nv)
	if err != nil{
		log.Panic(err)
	}
	fmt.Printf("收到来自%d节点的newview\n", nv.View)
	//fmt.Println(nv)
	//1、签名合法
	vcHash := sha256.Sum256([]byte(nv.ViewChangeSet))
	ppHash := sha256.Sum256([]byte(nv.PrePrepareSet))
	MessageNodePubKey := p.node.getPubKey(nv.View % nodeCount)
	digestByte := append(vcHash[:],ppHash[:]...)
	if nv.View != p.view {
		fmt.Println("发送节点不是主节点，拒绝接收newView")
	}else if !p.node.RsaVerySignWithSha256(digestByte, nv.Sign, MessageNodePubKey){
		fmt.Println("签名不合法，拒绝接收newView")
	}else{
		//2、计算和处理new-view
		//2-1、解码viewChange->map
		err := json.Unmarshal([]byte(nv.ViewChangeSet), &p.isViewChange)
		if err != nil{
			log.Println(err)
		}
		for _,content := range p.isViewChange{
			vc := new(ViewChange)
			err := json.Unmarshal([]byte(content), vc)
			if err!= nil{
				log.Panic(err)
			}
			//2-2、获取当前check_point
			if p.checkPoint < vc.SequenceID{
				p.checkViewChangePoint(vc)
			}
			//2-3、处理pre-prepare
			p.checkViewChangePre(vc.PrePrepareSet)
		}
		//3、处理pre-prepare
		prePrepareCompare := true
		prePrepareSet := make(map[int64]string)
		err = json.Unmarshal([]byte(nv.PrePrepareSet), &prePrepareSet)
		if err != nil{
			log.Println(err)
		}
		for eventId, prePre := range prePrepareSet{
			ppStr,err := json.Marshal(p.preparedMap[p.view][eventId])
			if err != nil{
				log.Println("err:",err)
			}
			if prePre != string(ppStr){
				prePrepareCompare = false
				break
			}
		}
		//4、检查收到数量足够与否->是
		count := 0
		for range p.isViewChange{
			count++
		}
		if count <= nodeCount * 2 / 3 - 1{
			fmt.Println("viewChange数量不足2f,拒绝接收newView:")
		}else if  prePrepareCompare != true{
			fmt.Println("prePrepare对比失败，拒绝接收newView")
		}else{
			//清空并重新存入pre-prepare数据
			DeletePrePrepare(p.db, p.checkPoint + p.checkSize * 2)
			for _,onePrePrepare := range p.preparedMap[p.view]{
				InsertPrePrepare(p.db, strconv.FormatInt(onePrePrepare.EventId, 10),
					strconv.Itoa(p.view), onePrePrepare.Digest, onePrePrepare.Message)
			}
			//5、关闭标志位
			p.ViewChangingFlag = false
			fmt.Println("本副节点已经更换主节点为",p.view)
		}
	}
}
